package com.demo.rocketmq.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;

/**
 * @author wangfengchen
 */
@Component
@Slf4j
public class DemoSequenceConsumerListener implements MessageListenerOrderly {

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
        if (CollectionUtils.isEmpty(list)) {
            log.error("接受到的消息为空，不处理，直接返回成功");
            return ConsumeOrderlyStatus.SUCCESS;
        }
        for (MessageExt messageExt : list) {
            int reconsume = messageExt.getReconsumeTimes();
            try {
                String msg = new String(messageExt.getBody());
                log.info("DemoSequenceConsumerListener消费者MQ收到消息, msg = {}，msgId = {}，topic = {}，tags = {}，keys = {}", msg, messageExt.getMsgId(), messageExt.getTopic(), messageExt.getTags(), messageExt.getKeys());
                //TODO 判断该消息是否重复消费
                //TODO 处理对应的业务逻辑
                // 如果没有return success ，consumer会重新消费该消息，直到return success
                return ConsumeOrderlyStatus.SUCCESS;
            } catch (Exception e) {
                log.error("DemoSequenceConsumerListener消费异常，e = {}", ExceptionUtils.getStackTrace(e));
                // 如果默认重试2次不成功，则记录，人工补偿
                if (reconsume >= 2) {
                    log.error("DemoSequenceConsumerListener消费重试次数达到" + reconsume + "，不再重试，返回成功，人工补偿");
                    //TODO 人工补偿
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            }
        }
        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
    }
}
